#ifdef _MSC_VER
#pragma warning(disable:4244 4267 4355) // for google protobuf
#endif

#include <iostream>
#include "LBServer.h"
#include "PBDispatcher.h"
#include "PBCodec.h"
#include "loadBalance.pb.h"
#include "ServerBootstrap.h"
#include "ClientBootstrap.h"
#include "DSTHMutex.h"
#include "Log.h"

using namespace std;

namespace dsth {

LBServer::LBServer(EventLoop* loop, const InetSocketAddress& addr, const InetSocketAddress& buddy, int heartbeatInterval) 
    :server_(new ServerBootstrap(loop, addr)),
    client_(new ClientBootstrap(loop, buddy, addr.ToHostPortFormat().c_str())),
    dispatcher_(new PBDispatcher()),
    codec_(new PBCodec(std::make_pair(dispatcher_.get(), &PBDispatcher::OnMessage),
       std::make_pair(this, &LBServer::OnErrorMsg))),
    status_(kBooting),
    loop_(loop),
    conn_buddy_(false),
    mutex_(new Mutex()), 
    master_hb_(),
    expired_(heartbeatInterval * 2) {
    dispatcher_->RegisterMessageCallback<Heartbeat>(std::make_pair(this, &LBServer::OnHeartbeat));
    dispatcher_->RegisterMessageCallback<QueryStatus>(std::make_pair(this, &LBServer::OnQueryStatus));
    dispatcher_->RegisterMessageCallback<NotifyStatus>(std::make_pair(this, &LBServer::OnNotifyStatus));
    server_->SetOnConnection(std::make_pair(this, &LBServer::OnConnect));
    server_->SetOnDisConnection(std::make_pair(this, &LBServer::OnDisConnect));
    server_->SetOnMessage(std::make_pair(codec_.get(), &PBCodec::OnMessage));

    client_->SetOnConnection(std::make_pair(this, &LBServer::OnConnectToBuddy));
    client_->SetOnDisConnection(std::make_pair(this, &LBServer::OnDisConnectFromBuddy));
    client_->SetOnRefused(std::make_pair(this, &LBServer::OnConnectRefused));
    client_->SetOnMessage(std::make_pair(codec_.get(), &PBCodec::OnMessage));
}

LBServer::~LBServer() {

}

void LBServer::OnConnectRefused() {
    assert(status_ != kSlaver);
    status_ = kMaster;
    cout << "I'm Master!" << endl; 
}

void LBServer::Run() {
    server_->Run();
    client_->Connect();
    loop_->RunAfterMilliSeconds(3000, std::make_pair(this, &LBServer::CheckBuddyResponse));
}

void LBServer::CheckBuddyResponse() {
    if (status_ == kBooting) {
        status_ = kMaster;
    }
}

void LBServer::OnConnect(Connection* conn) {
    cout << "A new connection from " << conn->GetPeerAdderess() << endl; 
}

void LBServer::OnDisConnect(Connection* conn) {
    cout << "A disconnection from " << conn->GetPeerAdderess() << endl; 
}

void LBServer::OnErrorMsg(Connection* conn, ChannelBuffer* buffer, Timestamp time) {
    cout << "A Error Msg, Fix Me! from " << conn->GetPeerAdderess() << endl; 
}

void LBServer::OnHeartbeat(const Connection* conn, const Heartbeat* beat, Timestamp time) {
    cout << "A Heartbeat From " << conn->GetPeerAdderess() << endl;
    if (beat->has_status()) {
        if (beat->status() == MASTER) {
            assert(status_ == kSlaver);
            MutexLock lock(mutex_);
            master_hb_ = time;
        }
        else {
            assert(false);
        }
    }
    else {

    }
}

void LBServer::OnQueryStatus(const Connection* conn, const QueryStatus* req, Timestamp time) {
    cout << "OnQueryStatus " << conn->GetPeerAdderess() << endl;
    NotifyStatus notify;
    // TODO: Need this?
    notify.set_host("1", 1);
    notify.set_port(2);
    notify.set_status(static_cast<ServerStatus>(status_));
    codec_->SendMessage(conn, &notify);
    
    // send heartbeat
    if (!conn_buddy_) {
        client_->Connect();
    }
    else {
        cout << "!!!has connected Fix Me!" << endl;
        loop_->RunPeriodMilliSeconds(10000, std::make_pair(this, &LBServer::HeartbeatPeriod));
    }
}

void LBServer::OnNotifyStatus(const Connection* conn, const NotifyStatus* notify, Timestamp time) {
    cout << "OnNotifyStatus, current status: "<< status_ << endl;
    // TODO: handle status transform
    Status buddy_status = static_cast<Status>(notify->status());
    if (status_ == kMaster && buddy_status == kMaster) {
        cout << "master vs master Fix Me!" << endl; 
        assert(false);
    }
    if (buddy_status == kMaster) {
        status_ = kSlaver;
        cout << "I'm Slaver!" << endl;
        // TODO: check Heartbeat
    }
    else {
        cout << "####Fix Me!" << endl; 
        status_ = kMaster;
        loop_->RunPeriodMilliSeconds(10000, std::make_pair(this, &LBServer::HeartbeatPeriod));
    }
}

void LBServer::HeartbeatPeriod() {
    assert(conn_buddy_);
    Connection* conn = client_->GetConnection();
    if (conn) {
        Heartbeat hb;
        hb.set_status(MASTER);
        codec_->SendMessage(conn, &hb);
    }
    else {
        // TODO: LogMe and remove timer!
        cout << "Fix Me conn == null!" << endl;
        assert(false);
    }
}

void LBServer::CheckMasterHeartbeat() {
    assert(status_ == kSlaver);
    MutexLock lock(mutex_);
    int64_t td = TimeDifference(Timestamp::GetNowTime(), master_hb_);
    if (expired_ < (td / Timestamp::kMicroSecondsPerSecond)) {
        
    }
}

void LBServer::OnConnectToBuddy(Connection* conn) {
    cout << "connect to buddy " << conn->GetPeerAdderess() << endl;
    conn_buddy_ = true;
    if (status_ == kBooting) {
        // ask buddy's status
        QueryStatus query;
        codec_->SendMessage(conn, &query);
    }
    else if(status_ == kMaster) {
        loop_->RunPeriodMilliSeconds(10000, std::make_pair(this, &LBServer::HeartbeatPeriod));
    }
    else {
        cout << "Fix Me! status_ == kSlaver but connect to master." << endl;
        assert(false);
    }
    
}

void LBServer::OnDisConnectFromBuddy(Connection* conn) {
    cout << "buddy disconnect " << conn->GetPeerAdderess() << endl;
    // TODO: handle buddy disconnect
}

} // namespace
